package day05;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Iterator;

import java.util.Iterator;

public class Text {
    public static void main(String[] args) throws Exception {
        OutputTag<String> late = new OutputTag<>("late", BasicTypeInfo.STRING_TYPE_INFO);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//时间特性（语义） 默认 ProcessingTime
        //env.getConfig().setAutoWatermarkInterval(1000);//间歇性生成Watermark的时间 默认200ms获取一次
        SingleOutputStreamOperator<String> ds = env.addSource(new SourceFunction<String>() {
            Boolean flag = true;
            int num = 1;
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (flag) {//当前时间作为事件时间
                    long processing = System.currentTimeMillis();
                    if (num % 2 == 0) {
                        processing -= 2000;
                    }
                    String line = processing + "\thainiu\t" + num;
                    System.out.println("source:" + line);
                    ctx.collectWithTimestamp(line, processing);
                    num++;
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
            @Override//watermark 为eventime - 1ms 认为（eventime - 1ms）及以前的人都来了
            public long extractAscendingTimestamp(String element) {
                return Long.valueOf(element.split("\t")[0]);
            }
        })
                .keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return value.split("\t")[1];
            }
        })
                .timeWindow(Time.seconds(2)) //窗口大小2s
                .allowedLateness(Time.seconds(2))//最大窗口延迟时间 如果错过窗口还能等2s
                .sideOutputLateData(late)

                .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        System.out.println("subtask:" + getRuntimeContext().getIndexOfThisSubtask() +
                                ",start:" + context.window().getStart() +
                                ",end:" + context.window().getEnd() +
                                ",waterMarks:" + context.currentWatermark() +
                                ",currentTime:" + System.currentTimeMillis());

                        Iterator<String> iterator = elements.iterator();
                        for (; iterator.hasNext(); ) {
                            String next = iterator.next();
                            System.out.println("windows-->" + next);
                            out.collect("on time:" + next);
                        }
                    }
                });


        ds.getSideOutput(late).print("late");//侧输出 错过窗口的2s内侧输出
        ds.print("ontime");


        env.execute();
    }
}
